use std::sync::{Arc, Mutex};
use async_trait::async_trait;
use mongodb::bson::{Document, Bson};
use sqlx::{PgPool, Error};
use anyhow::{Result, anyhow};
use serde_json::{Value, to_value};

use crate::storage::storage_provider::{IStorage, IStorageExt, ICollection};

/// PostgreSQL存储实现
#[derive(Debug)]
pub struct PgStorage {
    pool: PgPool,
}

impl PgStorage {
    /// 创建新的PostgreSQL存储实例
    pub async fn new(connection_string: &str) -> Result<Self> {
        let pool = PgPool::connect(connection_string).await?;
        Ok(Self { pool })
    }
    
    /// 确保集合表存在
    async fn ensure_table_exists(&self, db_name: &str, table_name: &str) -> Result<()> {
        // 清理表名，防止SQL注入
        let safe_table_name = Self::sanitize_identifier(table_name)?;
        
        // 创建表（如果不存在）
        let create_table_sql = format!(
            r#"CREATE TABLE IF NOT EXISTS "{}"."{}" (
                _id VARCHAR PRIMARY KEY,
                data JSONB NOT NULL
            )"#, 
            db_name, safe_table_name
        );
        
        sqlx::query(&create_table_sql)
            .execute(&self.pool)
            .await?;
        
        Ok(())
    }
    
    /// 清理标识符
    fn sanitize_identifier(identifier: &str) -> Result<String> {
        // 验证标识符格式，防止SQL注入
        if !identifier.chars().all(|c| c.is_alphanumeric() || c == '_' || c == '-' || c == '.' ) {
            return Err(anyhow!("Invalid identifier: {}", identifier));
        }
        Ok(identifier.to_string())
    }
    
    /// 将BSON Document转换为JSONB
    fn document_to_jsonb(doc: &Document) -> Result<serde_json::Value> {
        let value = serde_json::to_value(doc)?;
        Ok(value)
    }
    
    /// 将JSONB转换为BSON Document
    fn jsonb_to_document(value: &Value) -> Result<Document> {
        let doc = mongodb::bson::to_document(value)?;
        Ok(doc)
    }
}

/// PostgreSQL集合实现
#[derive(Debug)]
struct PgCollection<T: serde::Serialize + serde::de::DeserializeOwned + Unpin + Send + Sync> {
    storage: Arc<PgStorage>,
    db_name: String,
    collection_name: String,
    _phantom: std::marker::PhantomData<T>,
}

impl<T: serde::Serialize + serde::de::DeserializeOwned + Unpin + Send + Sync> PgCollection<T> {
    fn new(storage: Arc<PgStorage>, db_name: &str, collection_name: &str) -> Self {
        Self {
            storage,
            db_name: db_name.to_string(),
            collection_name: collection_name.to_string(),
            _phantom: std::marker::PhantomData,
        }
    }
}

#[async_trait]
impl<T: serde::Serialize + serde::de::DeserializeOwned + Unpin + Send + Sync> ICollection<T> for PgCollection<T> {
    async fn insert_one(&self, document: T) -> Result<()> {
        self.storage.ensure_table_exists(&self.db_name, &self.collection_name).await?;
        
        let json_value = to_value(document)?;
        let _id = json_value.get("_id")
            .and_then(|id| id.as_str())
            .map(|s| s.to_string())
            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
        
        let insert_sql = format!(
            r#"INSERT INTO "{}"."{}" (_id, data) VALUES ($1, $2)"#, 
            self.db_name, self.collection_name
        );
        
        sqlx::query(&insert_sql)
            .bind(_id)
            .bind(json_value.to_string())
            .execute(&self.storage.pool)
            .await?;
        
        Ok(())
    }
    
    async fn insert_many(&self, documents: Vec<T>) -> Result<()> {
        self.storage.ensure_table_exists(&self.db_name, &self.collection_name).await?;
        
        let mut tx = self.storage.pool.begin().await?;
        
        for doc in documents {
            let json_value = to_value(doc)?;
            let _id = json_value.get("_id")
                .and_then(|id| id.as_str())
                .map(|s| s.to_string())
                .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
            
            let insert_sql = format!(
                r#"INSERT INTO "{}"."{}" (_id, data) VALUES ($1, $2)"#, 
                self.db_name, self.collection_name
            );
            
            sqlx::query(&insert_sql)
                .bind(_id)
                .bind(json_value.to_string())
                .execute(&mut *tx)
                .await?;
        }
        
        tx.commit().await?;
        Ok(())
    }
    
    async fn find(&self, _filter: Document) -> Result<Vec<T>> {
        // 简单实现，实际需要根据filter构建WHERE条件
        let select_sql = format!(
            r#"SELECT data FROM "{}"."{}""#, 
            self.db_name, self.collection_name
        );
        
        let results = sqlx::query_scalar::<_, String>(&select_sql)
            .fetch_all(&self.storage.pool)
            .await?;
        
        let mut items = Vec::new();
        for result in results {
            let json_value: Value = serde_json::from_str(&result)?;
            let item: T = serde_json::from_value(json_value)?;
            items.push(item);
        }
        
        Ok(items)
    }
    
    async fn find_one(&self, filter: Document) -> Result<Option<T>> {
        // 简单实现，假设filter包含_id
        if let Some(_id) = filter.get("_id") {
            let select_sql = format!(
                r#"SELECT data FROM "{}"."{}" WHERE _id = $1"#, 
                self.db_name, self.collection_name
            );
            
            if let Some(data) = sqlx::query_scalar::<_, String>(&select_sql)
                .bind(_id.to_string())
                .fetch_optional(&self.storage.pool)
                .await?
            {
                let json_value: Value = serde_json::from_str(&data)?;
                let item: T = serde_json::from_value(json_value)?;
                return Ok(Some(item));
            }
        }
        
        Ok(None)
    }
    
    async fn update_one(&self, filter: Document, update: Document) -> Result<()> {
        // 简单实现，假设filter包含_id
        if let Some(_id) = filter.get("_id") {
            let update_sql = format!(
                r#"UPDATE "{}"."{}" SET data = data || $1::jsonb WHERE _id = $2"#, 
                self.db_name, self.collection_name
            );
            
            let update_json = serde_json::to_string(&update)?;
            
            sqlx::query(&update_sql)
                .bind(update_json)
                .bind(_id.to_string())
                .execute(&self.storage.pool)
                .await?;
        }
        
        Ok(())
    }
    
    async fn delete_one(&self, filter: Document) -> Result<()> {
        // 简单实现，假设filter包含_id
        if let Some(_id) = filter.get("_id") {
            let delete_sql = format!(
                r#"DELETE FROM "{}"."{}" WHERE _id = $1"#,
                self.db_name, self.collection_name
            );
            
            sqlx::query(&delete_sql)
                .bind(_id.to_string())
                .execute(&self.storage.pool)
                .await?;
        }
        
        Ok(())
    }
    
    async fn update_many(&self, filter: Document, update: Document) -> Result<()> {
        // 简化实现，需要将filter转换为WHERE条件，update转换为SET子句
        // 这里只处理$set操作符的情况
        if let Some(set) = update.get("$set") {
            // 将set转换为Document
            let set_doc = set.as_document().ok_or_else(|| anyhow::anyhow!("Invalid $set value"))?;
            
            // 构建SET子句
            let mut set_clauses = Vec::new();
            let mut values = Vec::new();
            let mut param_index = 1;
            
            for (key, value) in set_doc.iter() {
                let set_clause = format!("data ->> '{}' = ${}", key, param_index);
                set_clauses.push(set_clause);
                values.push(value.to_string());
                param_index += 1;
            }
            
            let set_sql = set_clauses.join(", ");
            
            // 构建WHERE子句
            let where_clause = if let Some(_id) = filter.get("_id") {
                format!("data ->> '_id' = ${}", param_index)
            } else {
                "1=1".to_string()
            };
            
            // 构建完整的UPDATE语句
            let update_sql = format!(
                r#"UPDATE "{}"."{}" SET {} WHERE {};"#,
                self.db_name, self.collection_name, set_sql, where_clause
            );
            
            // 执行更新操作
            let mut query = sqlx::query(&update_sql);
            
            // 绑定SET参数
            for value in values {
                query = query.bind(value);
            }
            
            // 绑定WHERE参数
            if let Some(_id) = filter.get("_id") {
                query = query.bind(_id.to_string());
            }
            
            query.execute(&self.storage.pool).await?;
        }
        
        Ok(())
    }
    
    async fn aggregate(&self, pipeline: Vec<Document>) -> Result<Vec<Document>> {
        // 简化实现，实际需要将MongoDB聚合管道转换为PostgreSQL查询
        let select_sql = format!(
            r#"SELECT data FROM "{}"."{}""#, 
            self.db_name, self.collection_name
        );
        
        let results = sqlx::query_scalar::<_, String>(&select_sql)
            .fetch_all(&self.storage.pool)
            .await?;
        
        let mut docs = Vec::new();
        for result in results {
            let json_value: Value = serde_json::from_str(&result)?;
            let doc = PgStorage::jsonb_to_document(&json_value)?;
            docs.push(doc);
        }
        
        Ok(docs)
    }
}

#[async_trait]
impl IStorage for PgStorage {
    fn get_document_collection(&self, db_name: &str, collection_name: &str) -> Box<dyn ICollection<Document>> {
        Box::new(PgCollection::new(Arc::new(self.clone()), db_name, collection_name))
    }
    
    async fn run_command(&self, db_name: &str, _command: Document) -> Result<Document> {
        // 实现基本命令处理
        let response = Document::new();
        Ok(response)
    }
    
    async fn list_collections(&self, db_name: &str) -> Result<Vec<String>> {
        let query = r#"SELECT table_name FROM information_schema.tables WHERE table_schema = $1"#;
        
        let results = sqlx::query_scalar::<_, String>(query)
            .bind(db_name)
            .fetch_all(&self.pool)
            .await?;
        
        Ok(results)
    }
    
    async fn insert_one_document(&self, db_name: &str, collection_name: &str, document: Document) -> Result<()> {
        let collection: Box<dyn ICollection<Document>> = self.get_document_collection(db_name, collection_name);
        collection.insert_one(document).await
    }
    
    async fn insert_many_documents(&self, db_name: &str, collection_name: &str, documents: Vec<Document>) -> Result<()> {
        let collection: Box<dyn ICollection<Document>> = self.get_document_collection(db_name, collection_name);
        collection.insert_many(documents).await
    }
    
    async fn find_documents(&self, db_name: &str, collection_name: &str, filter: Document) -> Result<Vec<Document>> {
        let collection: Box<dyn ICollection<Document>> = self.get_document_collection(db_name, collection_name);
        collection.find(filter).await
    }
    
    async fn find_one_document(&self, db_name: &str, collection_name: &str, filter: Document) -> Result<Option<Document>> {
        let collection: Box<dyn ICollection<Document>> = self.get_document_collection(db_name, collection_name);
        collection.find_one(filter).await
    }
    
    async fn update_one(
        &self,
        db_name: &str,
        collection_name: &str,
        filter: Document,
        update: Document,
    ) -> Result<()>
    {
        let collection: Box<dyn ICollection<Document>> = self.get_document_collection(db_name, collection_name);
        collection.update_one(filter, update).await
    }
    
    async fn delete_one(
        &self,
        db_name: &str,
        collection_name: &str,
        filter: Document,
    ) -> Result<()> {
        let collection: Box<dyn ICollection<Document>> = self.get_document_collection(db_name, collection_name);
        collection.delete_one(filter).await
    }
    
    async fn update_many(
        &self,
        db_name: &str,
        collection_name: &str,
        filter: Document,
        update: Document,
    ) -> Result<()> {
        let collection: Box<dyn ICollection<Document>> = self.get_document_collection(db_name, collection_name);
        collection.update_many(filter, update).await
    }
    
    async fn aggregate(
        &self,
        db_name: &str,
        collection_name: &str,
        pipeline: Vec<Document>,
    ) -> Result<Vec<Document>> {
        let collection: Box<dyn ICollection<Document>> = self.get_document_collection(db_name, collection_name);
        collection.aggregate(pipeline).await
    }
}

#[async_trait]
impl IStorageExt for PgStorage {
    fn get_collection<T>(&self, db_name: &str, collection_name: &str) -> Box<dyn ICollection<T>>
    where
        T: serde::Serialize + serde::de::DeserializeOwned + Unpin + Send + Sync + 'static
    {
        Box::new(PgCollection::new(Arc::new(self.clone()), db_name, collection_name))
    }
    
    async fn insert_one<T>(&self, db_name: &str, collection_name: &str, document: T) -> Result<()>
    where
        T: serde::Serialize + serde::de::DeserializeOwned + Unpin + Send + Sync + 'static
    {
        let collection: Box<dyn ICollection<T>> = self.get_collection(db_name, collection_name);
        collection.insert_one(document).await
    }
    
    async fn insert_many<T>(&self, db_name: &str, collection_name: &str, documents: Vec<T>) -> Result<()>
    where
        T: serde::Serialize + serde::de::DeserializeOwned + Unpin + Send + Sync + 'static
    {
        let collection: Box<dyn ICollection<T>> = self.get_collection(db_name, collection_name);
        collection.insert_many(documents).await
    }
    
    async fn find<T>(&self, db_name: &str, collection_name: &str, filter: Document) -> Result<Vec<T>>
    where
        T: serde::Serialize + for<'de> serde::Deserialize<'de> + Unpin + Send + Sync + 'static
    {
        let collection: Box<dyn ICollection<T>> = self.get_collection(db_name, collection_name);
        collection.find(filter).await
    }
    
    async fn find_one<T>(&self, db_name: &str, collection_name: &str, filter: Document) -> Result<Option<T>>
    where
        T: serde::Serialize + for<'de> serde::Deserialize<'de> + Unpin + Send + Sync + 'static
    {
        let collection: Box<dyn ICollection<T>> = self.get_collection(db_name, collection_name);
        collection.find_one(filter).await
    }
}

impl Clone for PgStorage {
    fn clone(&self) -> Self {
        Self {
            pool: self.pool.clone(),
        }
    }
}